package com.atguigu.edu.realtime220815.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime220815.bean.TableProcess;
import com.atguigu.edu.realtime220815.common.EduConfig;
import com.atguigu.edu.realtime220815.util.PhoenixUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.*;
import java.util.HashMap;
import java.util.Map;

/**
 * @Classname DimTableProcessFunction
 * @Description TODO
 * @Date 2023/2/15 20:12
 * @Created by lzx
 */
public class DimTableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private Map<String,TableProcess> configMap = new HashMap<>();


    @Override
    public void open(Configuration parameters) throws Exception {
        Class.forName("com.mysql.cj.jdbc.Driver");
        //建立连接
        Connection conn = DriverManager.getConnection("jdbc:mysql://hadoop100:3306/edu_config?" +
                "user=root&password=000000&useUnicode=true&characterEncoding=utf8&serverTimeZone=Asia/Shanghai&useSSL=false");
        //获取数据库操作对象
        String sql = "select * from edu_config.table_process where sink_type='dim'";
        PreparedStatement ps = conn.prepareStatement(sql);
        //执行SQL语句
        ResultSet rs = ps.executeQuery();
        //处理结果集
        ResultSetMetaData metaData = rs.getMetaData();
        while (rs.next()){
            JSONObject jsonObj = new JSONObject();
            for (int i = 1; i <= metaData.getColumnCount(); i++) {
                String columnName = metaData.getColumnName(i);
                Object columnValue = rs.getObject(i);
                jsonObj.put(columnName,columnValue);
            }
            TableProcess tableProcess = JSON.toJavaObject(jsonObj, TableProcess.class);
            configMap.put(tableProcess.getSourceTable(),tableProcess);
        }
        //释放资源
        rs.close();
        ps.close();
        conn.close();
    }
    private MapStateDescriptor<String , TableProcess> mapStateDescriptor = null;

    public DimTableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    //主流
    @Override
    public void processElement(JSONObject value, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        String database = value.getString("database");
        String table = value.getString("table");
        TableProcess tableProcess = null;
        if("edu".equals(database) && ((tableProcess = broadcastState.get(table))!=null) || (tableProcess = configMap.get(table))!=null ){
            //说明此条数据为维度数据
            String sinkColumns = tableProcess.getSinkColumns();
            JSONObject data = value.getJSONObject("data");

            //将test_question_info表中的question_txt字段中的 ' 换成  \' , 因为在拼接Phoenix的插入语句时单引号需要转义
            if(table.equals("test_question_info")){
                String questionTxt = data.getString("question_txt");
                String replaceTxt = questionTxt.replaceAll("'", "\\\\'");
                data.put("question_txt",replaceTxt);

            }

            //封装一个方法,用来过滤不用的字段
            filterColumns(data,sinkColumns);
            //过滤结束后,还需要将配置中的sink_table这个字段加到jsonObject中,不然的话发送到下游的时候还不知道这条数据
            //要发送到phoenix的哪张表
            data.put("sink_table",tableProcess.getSinkTable());
            //还要将操作类型也加上,要不然的话组织sql语句时不知道是删除还是别的类型
            data.put("type",value.getString("type"));
            //发往下游
            out.collect(data);
        }
    }
    private static void filterColumns(JSONObject jsonObject , String sinkColumns){
        String[] columns = sinkColumns.split(",");
        HashMap<String, String> tempMap = new HashMap<>();
        for (int i = 0; i < columns.length; i++) {
            String string = jsonObject.getString(columns[i]);
            if(!StringUtils.isEmpty(string)){
                tempMap.put(columns[i],string);
            }
        }
        jsonObject.clear();
        jsonObject.putAll(tempMap);
    }

    //配置流
    @Override
    public void processBroadcastElement(String value, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        JSONObject jsonObject = JSON.parseObject(value);
        String op = jsonObject.getString("op");
        //广播状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        if("d".equals(op)){
            String sourceTable = jsonObject.getJSONObject("before").getString("source_table");
            broadcastState.remove(sourceTable);
            configMap.remove(sourceTable);
        }else {
            JSONObject after = jsonObject.getJSONObject("after");
            String sourceTable = after.getString("source_table");
            String sourceType = after.getString("source_type");
            String sinkTable = after.getString("sink_table");
            String sinkType = after.getString("sink_type");
            String sinkColumns = after.getString("sink_columns");
            String sinkPk = after.getString("sink_pk");
            String sinkExtend = after.getString("sink_extend");
            TableProcess tableProcess = new TableProcess(sourceTable, sourceType, sinkTable, sinkType, sinkColumns, sinkPk, sinkExtend);
            broadcastState.put(sourceTable,tableProcess);
            createHbaseTable(sinkTable,sinkColumns,sinkPk,sinkExtend);
        }
    }

    private static void createHbaseTable(String sinkTable , String sinkColumns , String sinkPk ,String sinkExtend )  {
        if(sinkPk == null){
            sinkPk = "id";
        }
        if(sinkExtend == null){
            sinkExtend = "";
        }
        StringBuilder builder = new StringBuilder();
        builder.append("create table if not exists ").append(EduConfig.PHOENIX_NAMESPACE+"."+sinkTable+"( ") ;
        String[] split = sinkColumns.split(",");
        for (int i = 0; i < split.length; i++) {
            builder.append(split[i]+" varchar(4001) ");
            if(split[i].equals(sinkPk)){
                builder.append("primary key ");
            }
            if(i!=split.length-1){
                builder.append(" , ");
            }
        }
        builder.append(" ) "+sinkExtend);
        PhoenixUtils.executorSql(builder.toString());
    }
}
